package com.shujia.spark

import org.apache.spark.{SparkConf, SparkContext}

object Demo11Student {
  def main(args: Array[String]): Unit = {

    /**
      *
      * 1、统计年级排名前十学生各科的分数 [学号,学生姓名，学生班级，科目名，分数]
      *
      */

    val conf = new SparkConf().setAppName("Demo11Student").setMaster("local[4]")
    val sc = new SparkContext(conf)

    val student = sc.textFile("spark/data/students.txt")
    val score = sc.textFile("spark/data/score.txt")
    val cource = sc.textFile("spark/data/cource.txt")


    //计算学生总分
    val top10 = score.map(line => {
      val split = line.split(",")
      val id = split(0)
      val s = split(2).toInt
      (id, s)
    }).reduceByKey((x, y) => x + y)
      .sortBy(kv => kv._2, false) //总分倒叙排序‘
      .take(10)

    //取出前10学生的分数
    val top10Score = score.filter(line => {
      val id = line.split(",")(0)
      top10.map(_._1).contains(id)
    })

    //将top的RD转换成kv格式，关联学生表
    val top10KV = top10Score.map(line => {
      val id = line.split(",")(0)
      (id, line)
    })

    val studentKV = student.map(line => {
      val id = line.split(",")(0)
      (id, line)
    })

    //关联学生表
    val stuScoKV = top10KV.leftOuterJoin(studentKV).map(kv => {
      val id = kv._1
      val score = kv._2._1
      val scSplit = score.split(",")
      //科目编号
      val couId = scSplit(1)

      val sco = scSplit(2) //分数

      val stuInfo = kv._2._2

      val stu = stuInfo match {
        case Some(s) => s
        case None => "null,null,null,null,null"
      }

      val stuSplit = stu.split(",")
      val name = stuSplit(1)
      val clazz = stuSplit(4)

      //val v = id + "," + name + "," + clazz + "," + sco

      val v = s"${id},$name,$clazz,$sco"

      //科目编号作为k   去关联科目表
      (couId, v)
    })


    val couKV = cource.map(line => {
      val split = line.split(",")
      val id = split(0)
      val name = split(1)
      (id, name)
    })


    //关联科目表

    val resultRDD = stuScoKV.join(couKV).map(kv => {
      val i = kv._2
      val stuInfo = i._1
      val couName = i._2

      s"$stuInfo,$couName"
    })



    //2、统计总分大于年级平均分的学生 [学号，姓名，班级，总分]

    val sumSco = score.map(line => {
      val split = line.split(",")
      val id = split(0)
      val s = split(2).toInt
      (id, s)
    }).reduceByKey((x, y) => x + y)


    //计算平均分
    val avgSco = sumSco.map(_._2).reduce((x, y) => x + y) / sumSco.count().toDouble


    println(s"年级平均分：$avgSco")


    //取出总分大于年级平均分的学生
    val topNSco = sumSco.filter(kv => kv._2 > avgSco)


    //关联学生表
    topNSco.join(studentKV).map(kv => {
      val id = kv._1
      val sumSco = kv._2._1
      val stuInfo = kv._2._2
      val split = stuInfo.split(",")
      val name = split(1)
      val clazz = split(4)

      s"$id,$name,$clazz,$sumSco"
    }).foreach(println)


  }
}
